package com.sunny.service.v2;

import com.sunny.service.registry.IRegistryService;
import com.sunny.service.registry.ZKRegistryService;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Sunny
 * @date 2020/2/6
 */
@Component
public class CustomRpcServer implements ApplicationContextAware, InitializingBean {

    private Map<String, Object> serviceBeanMap = new ConcurrentHashMap<>();
    private IRegistryService registryService = new ZKRegistryService();

    private int port;

    public CustomRpcServer(int port) {
        this.port = port;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        NioEventLoopGroup boss = new NioEventLoopGroup();
        NioEventLoopGroup work = new NioEventLoopGroup();

        ServerBootstrap server = new ServerBootstrap();
        server.group(boss, work)
                .option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();

                        pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
                        pipeline.addLast(new ObjectEncoder());
                        pipeline.addLast(new CustomProcessHandler(serviceBeanMap));
                    }
                });

        ChannelFuture future = server.bind(port).sync();
        future.channel().closeFuture().sync();
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        //容器启动后获取需要发布的类
        Map<String, Object> beanMaps = applicationContext.getBeansWithAnnotation(RpcServer.class);
        if (!beanMaps.isEmpty()) {
            for (Object serviceBean : beanMaps.values()) {
                RpcServer rpcServer = serviceBean.getClass().getAnnotation(RpcServer.class);
                String serviceName = rpcServer.value().getName();
                String version = rpcServer.version();
                if (!StringUtils.isEmpty(version)) {
                    serviceName += "-" + version;
                }
                serviceBeanMap.put(serviceName, serviceBean);

                try {
                    InetAddress address = InetAddress.getLocalHost();
                    registryService.registry(serviceName, address.getHostAddress() + ":" + port);
                } catch (UnknownHostException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
